/**
* Author: fulishang
* Create Time  : 2016年8月5日,上午12:36:28
* Modify Time :
* Desc  : 
*/
package com.sam.si.dc.transformer;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.InputStreamReader;
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.log4j.Logger;
import org.springframework.integration.file.transformer.AbstractFilePayloadTransformer;
import org.springframework.util.Assert;

import com.sam.si.dc.metadata.Field;
import com.sam.si.dc.metadata.FileCatalog;

public class FileTransformer extends AbstractFilePayloadTransformer<List<Map<Object, Object>>> {

	private static final Logger LOGGER = Logger.getLogger(FileTransformer.class);

	private volatile Charset charset = Charset.defaultCharset();

	private Map<String, FileCatalog> fileCatalogs = new HashMap<>();
	private Map<String, List<Field>> fields = new HashMap<>();
	
	public FileTransformer(){}
	
	public FileTransformer(Map<String, FileCatalog> fileCatalogs, Map<String, List<Field>> fields) {
		this.fileCatalogs = fileCatalogs;
		this.fields = fields;
	}
	/**
	 * Set the charset name to use when copying the File to a String.
	 *
	 * @param charset
	 *            The charset.
	 */
	
	public void setCharset(String charset) {
		Assert.notNull(charset, "charset must not be null");
		Assert.isTrue(Charset.isSupported(charset), "Charset '" + charset + "' is not supported.");
		this.charset = Charset.forName(charset);
	}

	@Override
	protected List<Map<Object, Object>> transformFile(File file) throws Exception {
		List<Map<Object, Object>> records = new ArrayList<>();
		String invalidRowPrefix = "";
		boolean checkInvalidRow = false;
		boolean ignore = false;
		String[] invalideRowContent;
		FileCatalog fileCatalog = fileCatalogs.get(file.getName().substring(0, file.getName().lastIndexOf('.')));
		
		if (fileCatalog != null) {
			LOGGER.info("File catalog exist -- " + fileCatalog.getId());
			if (fileCatalog.getInvalidRow() != null) {
				checkInvalidRow = true;
				invalidRowPrefix = fileCatalog.getInvalidRow();
				LOGGER.info("invalid row prefix " + invalidRowPrefix);				
			}
		} else {
			throw new Exception("File catalog not found");
		}
		
		BufferedReader br = new BufferedReader(new InputStreamReader(new FileInputStream(file), this.charset));

		//get the field config by the catalog id
		List<Field> listFields = fields.get(fileCatalog.getId()); 
		if (checkInvalidRow) {
			invalideRowContent = invalidRowPrefix.split(",");
			
			for (String line = br.readLine(); line != null; line = br.readLine()) {
				for (int i=0; i<invalideRowContent.length; i++) {
					if (line.startsWith(invalideRowContent[i].trim())) {
						LOGGER.info("Ignore line starts with " + invalideRowContent[i]);
						ignore = true;
					} 
				}
				if (!ignore) {
					records.add(this.createRecords(line, fileCatalog, listFields));
				}
				//reset the identifier
				ignore = false;
			}
		} else {
			for (String line = br.readLine(); line != null; line = br.readLine()) {
				records.add(this.createRecords(line, fileCatalog, listFields));
			}
		}
		br.close();
		return records;
	}
	
	/**
	 * 
	 * @param fieldId
	 * @return check if have the config
	 */
	public boolean checkField(String fieldId) {
		boolean exist = false;
		for (Map.Entry<String, List<Field>> map : fields.entrySet()) {
			if (map.getKey().equals(fieldId))
				exist = true;
		}
		return exist;
	}

	/**
	 * 
	 * @param line
	 * @param fileCatalog
	 * @param listFields
	 * @return map <fieldname, fieldvalue>
	 */
	public Map<Object, Object> createRecords(String line, FileCatalog fileCatalog, List<Field> listFields) {
			
		String[] values = line.split(fileCatalog.getContentSpliter());
		Map<Object, Object> map = new HashMap<>();
		for (Field field : listFields) {
			map.put(field.getFieldName(), values[field.getSequence()]);
		}	
		
		return map;
		
	}
	
}
